package org.niit.service

import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.Answer
import org.niit.common.TService

/**
 * Date:2025/6/11
 * Author：Ys
 * Description:
 */
class EDUDataService extends TService[DStream[Answer]]{





  override def dataAnalysis(data: DStream[Answer]): Unit ={

    hotQuestionTop(data)
    gradeQuestionTop10(data)
    hotQuestionWithSubject(data)
    studentWithQuestionTop10(data)
    subjectWithQuestionTop10(data)
    studentMinScoreTop10(data)
  }
   //需求一：实时统计Top10热点题   --->某题出现的次数
  def hotQuestionTop(data: DStream[Answer]): Unit = {
    //1. 将数据进行转换 提取数据 组成键值对 (题目ID,1)
    val mapDS: DStream[(String, Int)] = data.map(line => {
      (line.question_id, 1)
    })

    //2. 将数据进行聚合，将相同key的数据进行聚合，相同key的数据就是 (题目ID,1)
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)


    //3.将聚合后的数据进行降序排序，然后取前10
    // 取前10是最后一步，所以我们可以直接放到foreachRDD中，因为取前10后要将数据存入数据库中
    reduceDS.foreachRDD(rdd=>{

      val sortRdd: RDD[(String, Int)] = rdd.sortBy(_._2, false)
      val top10: Array[(String, Int)] = sortRdd.take(10)

      //数据库的操作自行编写

      println("----实时统计Top10热点题----")
      top10.foreach(println)
    })



  }

  //需求二：实时统计答题最活跃的Top10年级  ---> 某年级出现次数最多
  def gradeQuestionTop10(data: DStream[Answer]): Unit = {

    data.map(line=>{
      (line.grade_id,1)
    }).reduceByKey(_+_).foreachRDD(rdd=>{
      val top10: Array[(String, Int)] = rdd.sortBy(_._2, false).take(10)
      println("----实时统计答题最活跃的Top10年级----")
      top10.foreach(println)
    })

  }

  //需求三：实时统计每个科目的Top10热点题   --> 某个科目中题 出现的次数
  def hotQuestionWithSubject(data: DStream[Answer]): Unit = {

    data.map( line=>{
      ( (line.subject_id,  line.question_id) ,1  )

    } )
      .reduceByKey( _ + _ )
      .foreachRDD(rdd=>{
       val top10 =  rdd.sortBy(_._2).take(10)
        println("----实时统计每个科目的Top10热点题----")
        top10.foreach(println)
      })
  }

  //需求四：实时统计每位学生得分最低的题目Top10  --> 某个学生某个题最低分数
  def studentWithQuestionTop10(data: DStream[Answer]): Unit = {

    data.map(line=>{
      ( (line.student_id,line.question_id),line.score )
    })
      .foreachRDD( rdd=>{
        val top10 = rdd.sortBy(_._2,true).take(10)
        println("----实时统计每位学生得分最低的题目Top10----")
        top10.foreach(println)
      } )

  }

  //需求五：实时统计每个科目的Top10的热点题目   --> 每个科目下 出现次数最多前10道
  //语文  判断题1题 20
  //数学  填空题3题 3
  //最终的输出结果 == ( 数学，（填空题3题，3）)
  def subjectWithQuestionTop10(data: DStream[Answer]): Unit = {
      //1. key :  (科目ID,题ID) value:1
      val reduceDS: DStream[((String, String), Int)] = data.map(line => {
        ((line.subject_id, line.question_id), 1)
      }).reduceByKey(_ + _)
    //2. key: 科目id value:(题ID,次数)
    val groupDS: DStream[(String, Iterable[(String, Int)])] = reduceDS.map {
      case ((subject_id, question_id), count) => {
        (subject_id, (question_id, count))
      }
    }.groupByKey()


    groupDS.foreachRDD(rdd=>{
      val top10 = rdd.mapValues(_.toList.sortBy(_._2).reverse.take(10))
      println("----实时统计每个科目的Top10热点题目----")
      top10.collect().foreach(println)
    })
  }
  //需求六：实时统计每个学生得分最低的前10道题，并带上所属题目
  //最终结果： ==》 （学生，[  (题目,分数),(题目,分数),(题目,分数),(题目,分数) ]）
  def studentMinScoreTop10(data: DStream[Answer]): Unit = {
    val groupDS: DStream[(String, Iterable[(String, Int)])] = data.map(line => {
      (line.student_id, (line.question_id, line.score))

    }).groupByKey()

    groupDS.foreachRDD(rdd=>{

     val top10 =  rdd.mapValues(iter=>{
        iter.toList.sortBy(_._2).take(10)
      })
      println("----实时统计每个学生得分最低的前10道题，并带上所属题目----")
      top10.collect().foreach(println)

    })

  }
}
